package com.bing;

import com.espertech.esper.client.*;
import org.junit.Test;

import java.util.Arrays;

/**
 * Created by Administrator on 2015/5/16.
 */
public class HelloEsper {

    @Test
    public void testHello() {

        Configuration configuration = new Configuration();//设置配置信息
        configuration.addEventType(MyEvent.class);  //添加事件类型


        EPServiceProvider provider = EPServiceProviderManager.getDefaultProvider(configuration); //创建引擎实例

        EPAdministrator epAdministrator = provider.getEPAdministrator();  //创建staement的管理接口实例
        EPStatement statement = epAdministrator.createEPL("select id,name from MyEvent(id>1).win:length_batch(3)");  //创建EPL查询语句实例,查询所有myEvent
        EventType eventType = epAdministrator.getConfiguration().getEventType("MyEvent");
        System.out.println("eventType:" + Arrays.asList(eventType.getPropertyNames()));
        statement.addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents) {
                System.out.println("update ");
                for (EventBean eventBean : newEvents) {
                    System.out.println("id" + eventBean.get("id") + "   name:" + eventBean.get("name"));
                }
            }
        });

        EPRuntime epRuntime = provider.getEPRuntime();
        epRuntime.sendEvent(new MyEvent(1, "bing"));
        epRuntime.sendEvent(new MyEvent(2, "bingou"));

        epRuntime.sendEvent(new MyEvent(4, "bingou4"));
        epRuntime.sendEvent(new MyEvent(5, "bingou5"));
        epRuntime.sendEvent(new MyEvent(7, "bingou6"));


    }

    @Test
    public void testContext() {
        EPServiceProvider provider = EPServiceProviderManager.getDefaultProvider();
        EPAdministrator administrator = provider.getEPAdministrator();
        EPRuntime epRuntime = provider.getEPRuntime();
        administrator.getConfiguration().addEventType(Esb.class);
        String epl1 = "create context esbtest partition by id,price from Esb";
        String epl2 = "context esbtest select context.id, context.name,context.key1,context.key2 from Esb";

        administrator.createEPL(epl1);
        EPStatement epl = administrator.createEPL(epl2);
        epl.addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents) {
                System.out.println("newEvent length" + newEvents.length);
                System.out.println(Esb.toString(newEvents[0]));
            }
        });

        Esb esb = new Esb(1, 20);
        epRuntime.sendEvent(esb);
        esb = new Esb(2, 20);
        epRuntime.sendEvent(esb);
        esb = new Esb(1, 10);
        epRuntime.sendEvent(esb);
        esb = new Esb(1, 20);
        epRuntime.sendEvent(esb);
    }

    @Test
    public void testCategoryContext() {
        EPServiceProvider provider = EPServiceProviderManager.getDefaultProvider();
        EPAdministrator administrator = provider.getEPAdministrator();
        EPRuntime runtime = provider.getEPRuntime();

        administrator.getConfiguration().addEventType(Esb.class);

        String ep1 = "create context esbCate group price<0 as low , group price>0 and price <50 as med, group price >100 as hig from Esb";
        String ep2 = "context esbCate select context.id, context.label,context.name ,price from Esb";

        administrator.createEPL(ep1);
        EPStatement epStatement = administrator.createEPL(ep2);

        epStatement.addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents) {
                System.out.println("context.Id: " + newEvents[0].get("id") + "  context.Name:" + newEvents[0].get("name") + " context.Label:" + newEvents[0].get("label") + " price:" + newEvents[0].get("price"));
            }
        });

        runtime.sendEvent(new Esb(-4, 1));
        runtime.sendEvent(new Esb(20, 2));
        runtime.sendEvent(new Esb(30, 4));
        runtime.sendEvent(new Esb(100, 15));
        runtime.sendEvent(new Esb(555, 34));
        runtime.sendEvent(new Esb(0, 1));
    }

    @Test
    public void testNoOverLappingContext() {
        EPServiceProvider epServiceProvider = EPServiceProviderManager.getDefaultProvider();
        EPAdministrator administrator = epServiceProvider.getEPAdministrator();
        EPRuntime runtime = epServiceProvider.getEPRuntime();

        String ep1 = "create context nol start st end `End`";   //通过` 转换关键字
        String ep2 = "context nol select * from Other";
        administrator.getConfiguration().addEventType("st", Start.class);  //重命名EventType,因为Start是关键字
        administrator.getConfiguration().addEventType(End.class);
        administrator.getConfiguration().addEventType(Other.class);

        administrator.createEPL(ep1);
        EPStatement statement = administrator.createEPL(ep2);
        statement.addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents) {
                System.out.println(newEvents[0].get("name"));
            }
        });
        runtime.sendEvent(new Start());
        runtime.sendEvent(new Other("can"));

        runtime.sendEvent(new End());

        runtime.sendEvent(new Other("can't"));

        runtime.sendEvent(new Start());
        runtime.sendEvent(new Other("can1"));
        runtime.sendEvent(new Other("can2"));
        runtime.sendEvent(new End());
    }

    /**
     * 测试esper cast功能
     */
    @Test
    public void testCast() {
        EPServiceProvider provider = EPServiceProviderManager.getDefaultProvider();
        EPAdministrator administrator = provider.getEPAdministrator();
        EPRuntime runtime = provider.getEPRuntime();
        administrator.getConfiguration().addEventType(Banana.class);
        String sp1 = "select avg(price) from Banana.win:length_batch(2)";
        String sp2 = "select cast(avg(price), int) as price from Banana.win:length_batch(2)";
        administrator.createEPL(sp1).addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] eventBeans, EventBean[] eventBeans1) {
                System.out.println("ave price:" + eventBeans[0].get("avg(price)") + "   class:" + eventBeans[0].get("avg(price)").getClass().getName());
            }
        });
        administrator.createEPL(sp2).addListener(new UpdateListener() {
            @Override
            public void update(EventBean[] eventBeans, EventBean[] eventBeans1) {
                System.out.println("ave price:" + eventBeans[0].get("price") + "   class:" + eventBeans[0].get("price").getClass().getName());
            }
        });

        runtime.sendEvent(new Banana(1));
        runtime.sendEvent(new Banana(2));
    }

    /**
     * 测试方法调用
     */

    @Test
    public void testInvoke() {
        EPServiceProvider provider=EPServiceProviderManager.getDefaultProvider();
        EPAdministrator administrator = provider.getEPAdministrator();
        EPRuntime runtime = provider.getEPRuntime();
        administrator.getConfiguration().addEventType(ComputeArea.class);
        administrator.getConfiguration().addImport(ComputeArea.class);

        String sp1="select r.getArea() as area from ComputeArea as r";
        String sp2="@Name(\"sp2\") select r.getArea(r.width,r.length) as area from ComputeArea as r";
        String sp3="select ComputeArea.getSarea(r.width,r.length) as area from ComputeArea as r";

        UpdateListener listener = new UpdateListener() {
            @Override
            public void update(EventBean[] newEvents, EventBean[] oldEvents) {
                System.out.println(newEvents[0].get("area"));
            }
        };
        administrator.createEPL(sp1).addListener(listener);
        EPStatement ep2 = administrator.createEPL(sp2);
        System.out.println(ep2.getName());
        ep2.addListener(listener);
        administrator.createEPL(sp3).addListener(listener);

        runtime.sendEvent(new ComputeArea(10,25));
    }
}
